package com.yuhang.wordcount;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

// 批处理word count
public class WordCount {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        String inputPath = "..\\Flink_java\\learnning\\src\\main\\resources\\hello.txt";

        //DataSource 是 dataSet的子类，点击源码
        DataSource<String> dataSource = env.readTextFile(inputPath);

        //返回的就是二元组的（word,1)
        AggregateOperator<Tuple2<String, Integer>> resultData = dataSource.flatMap(new MyFlatMapFunction())
                .groupBy(0)  //这里可以用数字表示数据的位置,其他重载方法可以用string类型
                .sum(1);      //这里数据同理表示数据的位置，

        resultData.print();

    }

    public static class MyFlatMapFunction implements FlatMapFunction<String, Tuple2<String,Integer>>{

        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {

            //TODO scala语法更快，更简洁，这里只能循环遍历
            String[] datas = value.split(" ");

            for(String data: datas){
                //注意方法没有返回值，那么Out方法就是一个迭代器，收集结果的，所以输出用out
                out.collect(new Tuple2<>(data,1));
            }

        }
    }
}
